package com.atguigu.app

import java.util

import com.alibaba.fastjson.JSON
import com.atguigu.bean.{OrderDetail, OrderInfo, SaleDetail, UserInfo}
import com.atguigu.constants.GmallConstants
import com.atguigu.utils.{MyEsUtil, MyKafkaUtil}
import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, InputDStream}
import redis.clients.jedis.Jedis
import org.json4s.native.Serialization

import collection.JavaConverters._

object SaleDetailApp {
  def main(args: Array[String]): Unit = {
    //1.创建sparkConf
    val sparkConf: SparkConf = new SparkConf().setAppName("SaleDetailApp").setMaster("local[*]")

    //2.创建StreamingContext
    val ssc: StreamingContext = new StreamingContext(sparkConf, Seconds(3))

    //3.消费kafka数据
    val orderInfoKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER, ssc)

    val orderDetailKafkaDStream: InputDStream[ConsumerRecord[String, String]] = MyKafkaUtil.getKafkaStream(GmallConstants.KAFKA_TOPIC_ORDER_DETAIL, ssc)


    //4.将json字符串数据转为样例类，并补全字段,同时将数据转为k，v类型，key是两条流的连接条件
    val orderInfoDStream = orderInfoKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderInfo: OrderInfo = JSON.parseObject(record.value(), classOf[OrderInfo])

        //补全字段
        orderInfo.create_date = orderInfo.create_time.split(" ")(0)
        orderInfo.create_hour = orderInfo.create_time.split(" ")(1).split(":")(0)

        (orderInfo.id, orderInfo)
      })
    })

    val orderDetailDStream = orderDetailKafkaDStream.mapPartitions(partition => {
      partition.map(record => {
        val orderDetail: OrderDetail = JSON.parseObject(record.value(), classOf[OrderDetail])
        (orderDetail.order_id, orderDetail)
      })
    })

    //5.对两条流做join操作
    //    val joinDStream: DStream[(String, (OrderInfo, OrderDetail))] = orderInfoDStream.join(orderDetailDStream)
    val joinDStream: DStream[(String, (Option[OrderInfo], Option[OrderDetail]))] = orderInfoDStream.fullOuterJoin(orderDetailDStream)

    //6.采用加缓存的方式处理因网络延迟所带来的数丢失问题
    val noUserInfoDStream: DStream[SaleDetail] = joinDStream.mapPartitions(partition => {
      //创建List集合用来存放结果数据集SaleDetail
      val details: util.ArrayList[SaleDetail] = new util.ArrayList[SaleDetail]()
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      implicit val formats = org.json4s.DefaultFormats
      //遍历每条数据
      partition.foreach { case (orderId, (infoOpt, detailOpt)) =>
        val orderInfoRedisKey: String = "orderInfo:" + orderId
        val orderDetailRedisKey: String = "orderDetail:" + orderId

        //a.判断订单表的数据是否存在
        if (infoOpt.isDefined) {
          //订单表数据存在
          //取出订单表数据
          val orderInfo: OrderInfo = infoOpt.get
          //a.2判断订单明细表是否存在
          if (detailOpt.isDefined) {
            //订单明细表数据存在
            //取出订单明细表数据
            val orderDetail: OrderDetail = detailOpt.get
            //a.3将两个表的数据组合成样例类
            val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
            details.add(detail)
          }

          //b.1将orderInfo数据转为Json字符串写入redis
          val orderInfoJson: String = Serialization.write(orderInfo)
          jedis.set(orderInfoRedisKey, orderInfoJson)
          //b.2将存入的orderInfo数据设置过期时间
          jedis.expire(orderInfoRedisKey, 30)

          //c.1查询redis中orderDetail数据
          //判断orderDetailrediskey是否存在
          if (jedis.exists(orderDetailRedisKey)) {
            //redis中存放的有能够关联上的orderDetail数据
            val orderDetaiJsonSet: util.Set[String] = jedis.smembers(orderDetailRedisKey)
            for (elem <- orderDetaiJsonSet.asScala) {
              //c.2将查询出来的Json字符串转为orderDetail样例类
              val orderDetail: OrderDetail = JSON.parseObject(elem, classOf[OrderDetail])
              //c.3将orderInfo和查询出来的orderDetail数据转为SaleDetail样例类
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              details.add(detail)
            }
          }
        } else {
          //orderInfo数据不在
          //d.1判断orderDetail数据是否存在
          if (detailOpt.isDefined) {
            //orderDetail数据在，则取出数据
            val orderDetail: OrderDetail = detailOpt.get

            //e.1查询redis中存放的OrderInfo数据
            //判断orderInforediskey是否存在
            if (jedis.exists(orderInfoRedisKey)) {
              //e.2取出redis中缓存的orderInfo数据
              val orderInfoJsonStr: String = jedis.get(orderInfoRedisKey)
              //将json字符串转为样例类
              val orderInfo: OrderInfo = JSON.parseObject(orderInfoJsonStr, classOf[OrderInfo])
              //e.3将orderDetail和查询出来的orderInfo数据转为SaleDetail样例类
              val detail: SaleDetail = new SaleDetail(orderInfo, orderDetail)
              details.add(detail)
            } else {
              //对方缓存中没有能够关联上的数据
              //f.1将自己（orderDetail）存入缓存，等待数据到来
              //将orderDetail转为Json字符串
              val orderDetailJosnStr: String = Serialization.write(orderDetail)
              //将json字符串写入redis，并设置过期时间
              jedis.sadd(orderDetailRedisKey, orderDetailJosnStr)
              jedis.expire(orderDetailRedisKey, 30)
            }
          }
        }
      }
      jedis.close()
      details.asScala.toIterator
    })

    //7.关联用户表数据
    val saleDetailDStream: DStream[SaleDetail] = noUserInfoDStream.mapPartitions(partition => {
      //创建redis连接
      val jedis: Jedis = new Jedis("hadoop102", 6379)
      val details: Iterator[SaleDetail] = partition.map(saleDetail => {
        //查询redis中用户表的数据，并关联
        val userInfoRedisKey: String = "userInfo:" + saleDetail.user_id
        val userInfoJsonStr: String = jedis.get(userInfoRedisKey)
        //将查询出来的Json字符串转为样例类
        val userInfo: UserInfo = JSON.parseObject(userInfoJsonStr, classOf[UserInfo])
        //关联用户表数据
        saleDetail.mergeUserInfo(userInfo)
        saleDetail
      })
      jedis.close()
      details
    })
    saleDetailDStream.print()

    //8.将关联后明细数据写入ES
    saleDetailDStream.foreachRDD(rdd=>{
      rdd.foreachPartition(partition=>{
        val list: List[(String, SaleDetail)] = partition.toList.map(saleDetail => {
          (saleDetail.order_detail_id, saleDetail)
        })
        MyEsUtil.insertBulk(GmallConstants.ES_SALEDETAIL_INDEX+"211227",list)
      })
    })

    ssc.start()
    ssc.awaitTermination()
  }
}
